#pragma once

#include "execution_defs.h"
#include "execution_manager.h"
#include "executor_abstract.h"
#include "index/ix.h"
#include "system/sm.h"
#include <memory>

class NestedLoopJoinExecutor : public AbstractExecutor {
private:
    std::unique_ptr<AbstractExecutor> left_;    
    std::unique_ptr<AbstractExecutor> right_;   
    size_t len_;                                
    std::vector<ColMeta> cols_;                 
    std::vector<Condition> fed_conds_;
    std::unique_ptr<RmRecord> result;
    std::vector<std::unique_ptr<RmRecord>> left_record;     
    std::vector<std::unique_ptr<RmRecord>> right_record;    
    std::vector<std::unique_ptr<RmRecord>>::iterator left_itrator;
    std::vector<std::unique_ptr<RmRecord>>::iterator right_itrator;   
    bool isend;

    void initialize() {
    // 将所有元组从左右子执行器中提取出来
        extractRecords(left_, left_record);
        extractRecords(right_, right_record);

        left_itrator = left_record.begin();
        right_itrator = right_record.begin();

        // 如果左右记录中的任何一个为空，设置 `isend` 为 true
        isend = left_itrator == left_record.end() || right_itrator == right_record.end();

        // 如果 `isend` 为 false，找到符合连接条件的元组并更新结果
        while (!isend && !evaluateConditions()) {
            moveNextIterator();
        }
        
        if (!isend) {
            updateResult();
        }
    }

    void extractRecords(std::unique_ptr<AbstractExecutor> &executor, std::vector<std::unique_ptr<RmRecord>> &record) {
        for (executor->beginTuple(); !executor->is_end(); executor->nextTuple()) {
            record.push_back(executor->Next());
        }
    }

    void moveNextIterator() {
        if (++right_itrator == right_record.end()) {
            right_itrator = right_record.begin();
            if (++left_itrator == left_record.end()) {
                isend = true;
            }
        }
    }

    bool evaluateConditions() {
        char *lbase = (*left_itrator)->data;
        char *rbase = (*right_itrator)->data;
        return std::all_of(fed_conds_.begin(), fed_conds_.end(), [&](Condition &cond) {
            assert(!cond.is_rhs_val);
            auto lvalue = convertValue(lbase, get_col_offset_lr(left_->cols(), cond.lhs_col));
            auto rvalue = convertValue(rbase, get_col_offset_lr(right_->cols(), cond.rhs_col));
            return cond.eval(lvalue, rvalue);
        });
    }

    void updateResult() {
        result = std::make_unique<RmRecord>(len_);
        memcpy(result->data, (*left_itrator)->data, left_->tupleLen());
        memcpy(result->data + left_->tupleLen(), (*right_itrator)->data, right_->tupleLen());
    }

    ColMeta get_col_offset_lr(const std::vector<ColMeta> &cols, const TabCol &target) {
        auto it = std::find_if(cols.begin(), cols.end(), [&target](const ColMeta &col) {
            return col.tab_name == target.tab_name && col.name == target.col_name;
        });
        assert(it != cols.end());
        return *it;
    }


public:

    NestedLoopJoinExecutor(std::unique_ptr<AbstractExecutor> left, 
                                                std::unique_ptr<AbstractExecutor> right,
                                                std::vector<Condition> conds) 
        : left_(std::move(left)), right_(std::move(right)), fed_conds_(std::move(conds)), isend(false) {
        len_ = left_->tupleLen() + right_->tupleLen();
        cols_ = left_->cols();
        auto right_cols = right_->cols();
        for (auto &col : right_cols) {
            col.offset += left_->tupleLen();
        }
        cols_.insert(cols_.end(), right_cols.begin(), right_cols.end());
    }

    void beginTuple() override {
        initialize();
    }

    void nextTuple() override {
        // Check if iterator is at the end
        assert(!is_end());

        // Step through the join conditions until either `isend` is true or a tuple satisfies the join conditions
        moveNextIterator();
        while (!isend && !evaluateConditions()) {
            moveNextIterator();
        }

        // If iterator is still not at the end, create a new `RmRecord` with the joined tuple data and set `result` to it
        if (!isend) {
            updateResult();
        }
    }

    [[nodiscard]] bool is_end() const override {
        return isend;
    }

    [[nodiscard]] size_t tupleLen() const override {
        return len_;
    }

    [[nodiscard]] const std::vector<ColMeta> &cols() const override {
        return cols_;
    }

    std::unique_ptr<RmRecord> Next() override {
        return std::move(result);
    }
    Rid &rid() override { return _abstract_rid; }
};

